package com.example.mybank.kafka;


import com.example.mybank.entity.Account;
import com.example.mybank.entity.Orders;
import com.example.mybank.entity.Statement;
import com.example.mybank.mapper.AccountMapper;
import com.example.mybank.mapper.OrdersMapper;
import com.example.mybank.mapper.ProductMapper;
import com.example.mybank.mapper.StatementMapper;
import com.example.mybank.tool.MyException;
import com.example.mybank.tool.RedisTool;
import com.example.mybank.tool.ResultCode;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.redisson.api.RLock;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.util.Date;
import java.util.Optional;
import java.util.concurrent.TimeUnit;

@Component
public class MyConsumer {

    private static final Logger LOGGER= LoggerFactory.getLogger("kafkaConsumer.class");

    @Autowired
    ObjectMapper objectMapper;

    @Autowired
    AccountMapper accountMapper;

    @Autowired
    OrdersMapper ordersMapper;

    @Autowired
    Account bankAccount;

    @Autowired
    StatementMapper statementMapper;

    @Autowired
    ProductMapper productMapper;

    @Transactional(rollbackFor = Exception.class)
    @KafkaListener(topics = {ConstString.TOPIC})
    public void onMessage1(ConsumerRecord<?, ?> consumerRecord, Consumer consumer) throws MyException, InterruptedException {
        Optional<?> optional = Optional.ofNullable(consumerRecord.value());
        if (!optional.isPresent()) {

            LOGGER.info("message={}, value={}",null,null);
        }

        String msg = (String)(optional.get());
        LOGGER.info("message:{}", msg);

        Long orderId = Long.valueOf(msg);
        Orders orders = ordersMapper.selectById(orderId);
        LOGGER.info("message={}, value={}",orderId,orders);

        RLock lock_bank = RedisTool.account_lock_get(bankAccount.getId());
        RLock lock_users = RedisTool.account_lock_get(orders.getAccount());



        try{
            boolean res = lock_bank.tryLock(10, TimeUnit.SECONDS);
            boolean res2 = lock_users.tryLock(10, TimeUnit.SECONDS);

            int count = accountMapper.pay(orders.getAccount(),orders.getMoney());
            if(count!=1){
                throw new MyException(ResultCode.Account_Pay_Fail);
            }
            System.out.println("扣钱成功："+" "+count+" "+orders);

            orders.setIs_deal(Boolean.TRUE);
            orders.setIs_pay(Boolean.TRUE);
            ordersMapper.updateById(orders);
            // 银行账户拿钱
            accountMapper.getMoney(bankAccount.getId(),orders.getMoney());

            // 新增流水记录
            Statement statement = new Statement(0L, orders.getMoney(), orders.getUsers(), orders.getAccount(), bankAccount.getId(), orders.getId(), new Date());
            statementMapper.insert(statement);

            // 增加已购买数量
            productMapper.incrementCount_sale(orders.getProduct());

            consumer.commitAsync();

        }finally {
            lock_bank.unlock();
            lock_users.unlock();

            orders.setIs_deal(Boolean.TRUE);
            ordersMapper.updateById(orders);

            RedisTool.product_incHashValue(orders.getProduct());

        }
    }


}